package top.atcra.rpc;

import cn.hutool.core.lang.Assert;
import cn.hutool.core.lang.Pair;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.AttributeKey;
import io.netty.util.ResourceLeakDetector;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import top.atcra.rpc.annotation.scanner.RPCAnnotationScanner;
import top.atcra.rpc.compress.CompressorType;
import top.atcra.rpc.config.AppConfig;
import top.atcra.rpc.config.ConfigLoader;
import top.atcra.rpc.discovery.register.ServiceRegistry;
import top.atcra.rpc.discovery.register.ServiceRegistryFactory;
import top.atcra.rpc.exception.NetworkException;
import top.atcra.rpc.protection.circuitBreaker.CircuitBreaker;
import top.atcra.rpc.protection.rateLimit.RateLimiter;
import top.atcra.rpc.protection.health.ServiceHealthDetector;
import top.atcra.rpc.protection.loadBalancer.LoadBalanceType;
import top.atcra.rpc.protection.loadBalancer.LoadBalancer;
import top.atcra.rpc.protection.loadBalancer.LoadBalancerFactory;
import top.atcra.rpc.serialization.SerializerType;
import top.atcra.rpc.server.shutdown.ShutdownHook;
import top.atcra.rpc.service.ImplementRegistry;
import top.atcra.rpc.service.ReferenceConfig;
import top.atcra.rpc.service.ServiceConfig;
import top.atcra.rpc.transport.config.ServerRPCChannelInitializer;
import top.atcra.rpc.transport.utils.NettyBootstrapFactory;
import top.atcra.rpc.transport.utils.RequestIdGenerator;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * RPC框架启动器
 *
 * @author atcra
 * @date 2025-3-22
 */
@Slf4j
public class RPCBootstrap {

    public enum Role {
        CLIENT,
        SERVER
    }

    @Getter
    private Role role;
    @Getter
    private ServiceRegistry serviceRegistry;
    AppConfig currentAppConfig = null;

    public RPCBootstrap(Role role) {
        this.role = role;
        onCommonConfigReload(ConfigLoader.getConfig());
//        ConfigLoader.getInstance().setOnConfigLoaded(this::onCommonConfigReload);
        if (role == Role.CLIENT) {
            onClientConfigReload(ConfigLoader.getConfig());
            // 客户端无需热重载
//            ConfigLoader.getInstance().setOnConfigLoaded(this::onClientConfigReload);
        }
    }


    private Void onCommonConfigReload(AppConfig config) {
        if (currentAppConfig == null || !currentAppConfig.getRegistryUrl().equals(config.getRegistryUrl())) {
            serviceRegistry = ServiceRegistryFactory.getRegistry(this, config.getRegistryUrl());
        }
        currentAppConfig = config;
        return null;
    }

    /**
     * 设置应用名称
     *
     * @param applicationName 应用名称
     * @return RPCBootstrap
     */
    public RPCBootstrap application(String applicationName) {
        ConfigLoader.getConfig().setAppName(applicationName);
        return this;
    }

    /**
     * 指定注册中心URL
     *
     * @param registryURL 示例：zookeeper://127.0.0.1:2181
     * @return RPCBootstrap
     */
    public RPCBootstrap registry(String registryURL) {
        ConfigLoader.getConfig().setRegistryUrl(registryURL);
        serviceRegistry = ServiceRegistryFactory.getRegistry(this, registryURL);
        return this;
    }

    /**
     * 指定序列化方法
     *
     * @param serializerName 序列化方法名称
     * @return RPCBootstrap
     */
    public RPCBootstrap serializer(String serializerName) {
        ConfigLoader.getConfig().setSerializer(SerializerType.valueOf(serializerName).getId());
        return this;
    }

    /**
     * 指定压缩方法
     *
     * @param compressorName 压缩方法名称
     * @return RPCBootstrap
     */
    public RPCBootstrap compressor(String compressorName) {
        ConfigLoader.getConfig().setCompressor(CompressorType.valueOf(compressorName).getId());
        return this;
    }

    /**
     * 关闭客户端/服务端
     */
    public void shutdown() {
        try {
            if (role == Role.CLIENT) {
                clientEventLoopGroup.shutdownGracefully().sync();
                log.info("客户端已关闭");
            } else {
                serverBossEventLoopGroup.shutdownGracefully().sync();
                serverWorkerEventLoopGroup.shutdownGracefully().sync();
                log.info("服务端已关闭");
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    // ====================================== 以下配置仅用于客户端 ======================================
    @Getter
    ServiceHealthDetector serviceHealthDetector;
    @Getter
    private Bootstrap nettyBootstrap; // 用于创建新连接
    private NioEventLoopGroup clientEventLoopGroup;

    // Client: ID生成器
    @Getter
    RequestIdGenerator requestIdGenerator;

    // Client: 连接的缓存
    public static Map<RPCBootstrap, Map<InetSocketAddress, Channel>> CHANNEL_CACHE;

    // Client: 利用TreeMap对channel的超时时间进行管理，key是超时时间，value是channel。这个对象只在服务健康检测线程中写入，其他线程读取
    public static TreeMap<Long, Channel> CHANNEL_TTL_CACHE;
    @Getter
    LoadBalancer loadBalancer;

    // Client: 为每一个ip配置一个断路器，熔断
    public static final Map<SocketAddress, CircuitBreaker> HOST_CIRCUIT_BREAKERS = new ConcurrentHashMap<>(16);

    // key是请求id，value是请求的Future对象和起始时间
//    public static final ConcurrentHashMap<Long, Pair<CompletableFuture<Object>, Long>> PENDING_CALLS = new ConcurrentHashMap<>(1024);

    // 一些Key
    public static final AttributeKey<ConcurrentHashMap<Long, Pair<CompletableFuture<Object>, Long>>>
            PENDING_CALLS_KEY = AttributeKey.valueOf("PENDING_CALLS");
    public static final AttributeKey<RequestIdGenerator>
            REQUEST_ID_GENERATOR_KEY = AttributeKey.newInstance("RequestIdGenerator");
    public static final AttributeKey<RPCBootstrap>
            RPC_BOOTSTRAP_KEY = AttributeKey.newInstance("RPCBootstrap");

    public Map<InetSocketAddress, Channel> getChannelCache() {
        return CHANNEL_CACHE.get(this);
    }

    public RPCBootstrap reference(ReferenceConfig<?> referenceConfig) {
        Assert.notNull(nettyBootstrap, "please call initClient() first");
        Assert.notNull(serviceRegistry, "please call registry() first");
        loadBalancer.initSelector(referenceConfig.getServiceName(), referenceConfig.getServiceGroup());
        referenceConfig.setBootstrap(this);
        return this;
    }

    /**
     * 启用全周期服务健康检查（默认启动闲时服务健康检查）
     *
     */
    public RPCBootstrap enableDurableHealthCheck() {
        serviceHealthDetector = new ServiceHealthDetector(this);
        loadBalancer.enableHealthCheck(serviceHealthDetector);
        return this;
    }

    /**
     * 创建一个新的连接，并设置上下文信息（含PENDING_CALLS等
     * @param address 目标地址
     * @return 新的连接
     */
    public synchronized Channel createChannel(SocketAddress address) {
        try {
            var channel= nettyBootstrap.connect(address).sync().channel();
            channel.attr(PENDING_CALLS_KEY)
                    .set(new ConcurrentHashMap<>(1024));
            channel.attr(REQUEST_ID_GENERATOR_KEY)
                    .set(requestIdGenerator);
            channel.attr(RPC_BOOTSTRAP_KEY).set(this);
            return  channel;
        } catch (Exception e) {
            log.error("获取通道时，发生异常:{}", e.getMessage());
            throw new NetworkException(e);
        }
    }

    private Void onClientConfigReload(AppConfig config) {
        synchronized (Role.CLIENT) {
            // ID-Generator
            if (requestIdGenerator == null || currentAppConfig.getClientId() != config.getClientId()) {
                requestIdGenerator = new RequestIdGenerator(config.getDataCenterId(), config.getClientId());
            }
            // 负载均衡策略
            if (loadBalancer == null || currentAppConfig.getLoadBalancer() != config.getLoadBalancer()) {
                if (LoadBalanceType.min_response_time == LoadBalanceType.valueOf(config.getLoadBalancer())) {
                    loadBalancer = LoadBalancerFactory.getMinResponseTimeLoadBalancer(serviceRegistry, CHANNEL_TTL_CACHE);
                } else {
                    loadBalancer = LoadBalancerFactory.getLoadBalancer(this.serviceRegistry, config.getLoadBalancer());
                }
            }
            // Netty
            var pair = NettyBootstrapFactory.getClientBootstrap();
            nettyBootstrap = pair.getKey();
            clientEventLoopGroup = pair.getValue();

            // Channel-Cache
            if (CHANNEL_CACHE == null) {
                CHANNEL_CACHE = new ConcurrentHashMap<>(8);
            }
            CHANNEL_CACHE.put(this, new ConcurrentHashMap<>(16));
            if (CHANNEL_TTL_CACHE == null) {
                CHANNEL_TTL_CACHE = new TreeMap<>();
            }
        }
        currentAppConfig = config;
        return null;
    }

    // ====================================== 以下配置仅用于服务方 ======================================


    // Server: 为每一个ip配置一个限流器
    public static final Map<SocketAddress, RateLimiter> RATE_LIMITERS = new ConcurrentHashMap<>(16);

    protected EventLoopGroup serverBossEventLoopGroup;
    protected EventLoopGroup serverWorkerEventLoopGroup;

    public RPCBootstrap subnet(String subnetPrefix) {
        ConfigLoader.getConfig().setSubnetPrefix(subnetPrefix);
        return this;
    }

    /**
     * 启动服务器
     */
    public void startServer() {
        role = Role.SERVER;
        Runtime.getRuntime().addShutdownHook(new ShutdownHook());
        // boss只负责处理请求，之后会将请求分发至worker
        serverBossEventLoopGroup = new NioEventLoopGroup(2);
        serverWorkerEventLoopGroup = new NioEventLoopGroup(8);
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap()
                    .group(serverBossEventLoopGroup, serverWorkerEventLoopGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ServerRPCChannelInitializer());

            log.info("正在启动服务器，监听端口：{}", ConfigLoader.getConfig().getPort());
            ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.ADVANCED);
            ChannelFuture channelFuture = serverBootstrap.bind(ConfigLoader.getConfig().getPort()).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("启动服务器时发生异常。", e);
        }
    }

    protected static final AtomicLong requestCount = new AtomicLong(0);

    public static void qpsCountIncrease() {
        requestCount.incrementAndGet();
    }

    public static void startQPSPrinterThread() {
        Thread thread = new Thread(() ->
                new Timer().scheduleAtFixedRate(new TimerTask() {
                    @Override
                    public void run() {
                        System.out.println("QPS is " + requestCount.get());
                        System.out.flush();
                        requestCount.set(0);
                    }
                }, 0, 1000),
                "rpc-QPS-Thread");
        thread.setDaemon(true);
        thread.start();
    }


    /**
     * 扫描并发布指定包下的所有服务实现类
     *
     * @param packageFullName 包名
     */
    public RPCBootstrap publish(String packageFullName) {
        RPCAnnotationScanner scanner = new RPCAnnotationScanner();
        List<ServiceConfig> services = scanner.serviceImplementScan(packageFullName);
        for (var s : services) {
            log.info("发布服务：{}({})，实现类为{}", s.getName(), s.getGroup(), s.getRef().getClass().getName());
            publish(s);
        }
        return this;
    }


    /**
     * 发布服务
     *
     * @param serviceConfigs 服务的定义类（可以有多个）
     */
    public RPCBootstrap publish(ServiceConfig... serviceConfigs) {
        for (ServiceConfig serviceConfig : serviceConfigs) {
            Assert.notNull(serviceConfig.getRef(), "服务实现类不能为空");
            serviceRegistry.register(serviceConfig);
            ImplementRegistry.register(serviceConfig);
        }
        return this;
    }


}
